package com.springboot.rocketmq.consumer;

import com.springboot.rocketmq.message.MessageListen;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * Description：消费者配置
 */
//@Configuration
public class RocketMQConsumer {
    /**
     * 日志
     */
    private static Log logger = LogFactory.getLog(RocketMQConsumer.class);

    @Autowired
    private MessageListen messageListen;

    @Value("${rocketmq.consumer.namesrvAddr}")
    private String namesrvAddr;

    @Value("${rocketmq.consumer.groupName}")
    private String groupName;

    @Value("${rocketmq.consumer.topic}")
    private String topic;

    @Value("${rocketmq.consumer.tag}")
    private String tag;

    @Value("${rocketmq.consumer.consumeThreadMin}")
    private int consumeThreadMin;

    @Value("${rocketmq.consumer.consumeThreadMax}")
    private int consumeThreadMax;

    @Bean(initMethod = "start", destroyMethod = "shutdown")
    public DefaultMQPushConsumer getRocketMQConsumer() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumeThreadMax(consumeThreadMax);
        consumer.setVipChannelEnabled(false);
        // 我们自己实现的监听类
        consumer.registerMessageListener(messageListen);
        try {
            consumer.subscribe(topic, tag);
            logger.info("================>消费者创建完成，ConsumerGroupName: " + groupName);
            logger.info("============>消费者监听开始,groupName:" + groupName + ",topic:" + topic);
        } catch (MQClientException e) {
            logger.error("消费者启动失败");
            e.printStackTrace();
        }
        return consumer;
    }

}
